#include "event/znx_tcp_proc.h"
#include "event/znx_worker_proc.h"


static int64_t      task = 10;


void znx_test_server_read_1(znx_event_t *event)
{
    char                    buf[64];
    znx_result_t            ret;
    znx_conn_t              *conn = event->data;

    ret = znx_conn_read(conn, buf, 60);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect read success, but got other. ret: %d", ret);
    }

    buf[ret] = '\0';
    ZNX_DEF_LOG_DEBUG(NULL, "server_1: read %s", buf);

    ret = znx_conn_write(conn, "11111", 5);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    znx_event_loop_remove_conn(conn);
    znx_conn_free(conn);
}


void znx_test_server_accept_handler_1(znx_event_t *event)
{
    znx_conn_t *server_conn = event->data;
    server_conn->read.handler = znx_test_server_read_1;
    znx_event_loop_start_io(server_conn);
    znx_event_loop_enable_read(server_conn);
}

void znx_test_server_read_2(znx_event_t *event)
{
    char                    buf[64];
    znx_result_t            ret;
    znx_conn_t              *conn = event->data;

    ret = znx_conn_read(conn, buf, 60);
    if (ret == ZNX_AGAIN) {
        znx_event_loop_enable_read(conn);
        return;
    }

    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect read success, but got other, ret: %d", ret);
    }

    buf[ret] = '\0';
    ZNX_DEF_LOG_DEBUG(NULL, "server_2: read %s", buf);

    ret = znx_conn_write(conn, "22222", 5);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    znx_event_loop_remove_conn(conn);
    znx_conn_free(conn);
}

void znx_test_server_accept_handler_2(znx_event_t *event)
{
    znx_conn_t *server_conn = event->data;
    server_conn->read.handler = znx_test_server_read_2;
    znx_event_loop_start_io(server_conn);
    znx_event_loop_enable_read(server_conn);
}


void znx_test_server_read_4(znx_event_t *event)
{
    char                    buf[64];
    znx_result_t            ret;
    znx_conn_t              *conn = event->data;

    ret = znx_conn_read(conn, buf, 60);
    if (ret == ZNX_AGAIN) {
        znx_event_loop_enable_read(conn);
        return;
    }

    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect read success, but got other. ret: %d", ret);
    }

    buf[ret] = '\0';
    ZNX_DEF_LOG_DEBUG(NULL, "server_4: read %s", buf);

    ret = znx_conn_write(conn, "44444", 5);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    znx_event_loop_remove_conn(conn);
    znx_conn_free(conn);
}


void znx_test_server_accept_handler_4(znx_event_t *event)
{
    znx_conn_t *server_conn = event->data;
    server_conn->read.handler = znx_test_server_read_4;
    znx_event_loop_start_io(server_conn);
    znx_event_loop_enable_read(server_conn);
}


void znx_test_client_connect_read_handler_1(znx_event_t *event)
{
    char                        buf[64];
    znx_result_t                ret;
    znx_conn_t                  *client_conn = event->data;

    do {
        ret = znx_conn_read(client_conn, buf, 60);
        if (ret > 0) {
            buf[ret] = '\0';
            ZNX_DEF_LOG_DEBUG(NULL, "client_1: read data: %s", buf);
        }
    } while (ret > 0);

    if (ret == ZNX_AGAIN) {
        znx_event_loop_enable_read(client_conn);
        return;
    }

    if (ret == 0) {
        ZNX_DEF_LOG_DEBUG(NULL, "client_1: server closed");
    }

    if (ret < 0) {
        ZNX_DEF_LOG_DEBUG(NULL, "client_1: read error, errno: %d", -ret);
    }

    znx_event_loop_remove_conn(client_conn);
    znx_conn_free(client_conn);
}


void znx_test_client_connect_handler_1(void *user_data, znx_conn_t *conn, znx_result_t res)
{
    if (res != ZNX_SUCCESS) {
        ZNX_DEF_LOG_FATAL(NULL, "expect connect success, but got other");
    }

    res = znx_conn_write(conn, "aaaaa", 5);
    if (res != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    conn->read.handler = znx_test_client_connect_read_handler_1;
    znx_event_loop_start_io(conn);
    znx_event_loop_enable_read(conn);
}


void znx_test_client_connect_read_handler_2(znx_event_t *event)
{
    char                        buf[64];
    znx_result_t                ret;
    znx_conn_t                  *client_conn = event->data;

    do {
        ret = znx_conn_read(client_conn, buf, 60);
        if (ret > 0) {
            buf[ret] = '\0';
            ZNX_DEF_LOG_DEBUG(NULL, "client_2: read data: %s", buf);
        }
    } while (ret > 0);

    if (ret == ZNX_AGAIN) {
        znx_event_loop_enable_read(client_conn);
        return;
    }

    if (ret == 0) {
        ZNX_DEF_LOG_DEBUG(NULL, "client_2: server closed");
    }

    if (ret < 0) {
        ZNX_DEF_LOG_ERROR(NULL, "client_2: read error, errno: %d", -ret);
    }

    znx_event_loop_remove_conn(client_conn);
    znx_conn_free(client_conn);
}


void znx_test_client_connect_handler_2(void *user_data, znx_conn_t *conn, znx_result_t res)
{
    if (res != ZNX_SUCCESS) {
        ZNX_DEF_LOG_FATAL(NULL, "expect connect success, but got other");
    }

    znx_result_t ret;
    ret = znx_conn_write(conn, "bbbbb", 5);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    conn->read.handler = znx_test_client_connect_read_handler_2;
    znx_event_loop_start_io(conn);
    znx_event_loop_enable_read(conn);
}


void znx_test_client_connect_handler_3(void *user_data, znx_conn_t *conn, znx_result_t res)
{
    if (res == ZNX_SUCCESS) {
        ZNX_DEF_LOG_FATAL(NULL, "expect connect failed, but got success");
    }

    znx_conn_free(conn);
}


void znx_test_client_connect_read_handler_4(znx_event_t *event)
{
    char                        buf[64];
    znx_result_t                ret;
    znx_conn_t                  *client_conn = event->data;

    do {
        ret = znx_conn_read(client_conn, buf, 60);
        if (ret > 0) {
            buf[ret] = '\0';
            ZNX_DEF_LOG_DEBUG(NULL, "client_4: read data: %s", buf);
        }
    } while (ret > 0);

    if (ret == ZNX_AGAIN) {
        znx_event_loop_enable_read(client_conn);
        return;
    }

    if (ret == 0) {
        ZNX_DEF_LOG_DEBUG(NULL, "client_4: server closed");
    }

    if (ret < 0) {
        ZNX_DEF_LOG_ERROR(NULL, "client_4: read error, errno: %d", -ret);
    }

    znx_event_loop_remove_conn(client_conn);
    znx_conn_free(client_conn);
}


void znx_test_client_connect_handler_4(void *user_data, znx_conn_t *conn, znx_result_t res)
{
    if (res != ZNX_SUCCESS) {
        ZNX_DEF_LOG_FATAL(NULL, "expect connect success, but got other");
    }

    znx_result_t    ret;
    ret = znx_conn_write(conn, "ddddd", 5);
    if (ret != 5) {
        ZNX_DEF_LOG_FATAL(NULL, "expect write success, but got other");
    }

    conn->read.handler = znx_test_client_connect_read_handler_4;
    znx_event_loop_start_io(conn);
    znx_event_loop_enable_read(conn);
}


void znx_test_client_handler(znx_dispatch_event_t *dispatch_event)
{
    int data = (int)(intptr_t)dispatch_event->data;
    int case_no = (int)(intptr_t)dispatch_event->save_data_1;
    znx_thread_free(dispatch_event);

    znx_address_t address;
    bzero(&address, sizeof(znx_address_t));
    znx_conn_t *conn = znx_thread_calloc(sizeof(znx_conn_t));
    znx_conn_init(conn, ZNX_CONN_EPOLL_ET);
    
    if (case_no == 1) {             // znx_test_server_accept_handler_1
        if (!znx_address_parser(
            (u_char *)"127.0.0.1:8889", strlen("127.0.0.1:8889"), &address, NULL))
        {
            ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8889");
        }
        znx_tcp_connect_setup(conn, &address, NULL, NULL, 10, znx_test_client_connect_handler_1, (void *)(intptr_t)data);
    } else if (case_no == 2) {      // znx_test_server_accept_handler_2
        if (!znx_address_parser(
            (u_char *)"127.0.0.1:8889", strlen("127.0.0.1:8889"), &address, NULL))
        {
            ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8889");
        }
        znx_tcp_connect_setup(conn, &address, NULL, NULL, 10, znx_test_client_connect_handler_2, (void *)(intptr_t)data);
    } else if (case_no == 3) {      // RST
        if (!znx_address_parser(
            (u_char *)"127.0.0.1:8889", strlen("127.0.0.1:8889"), &address, NULL))
        {
            ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8889");
        }
        znx_tcp_connect_setup(conn, &address, NULL, NULL, 10, znx_test_client_connect_handler_3, (void *)(intptr_t)data);
    } else {                        // znx_test_server_accept_handler_4
        if (!znx_address_parser(
            (u_char *)"127.0.0.1:8888", strlen("127.0.0.1:8888"), &address, NULL))
        {
            ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8888");
        }
        znx_tcp_connect_setup(conn, &address, NULL, NULL, 10, znx_test_client_connect_handler_4, (void *)(intptr_t)data);
    }
}


void znx_tcp_test(int case_no)
{
    znx_dispatch_event_t            *dispatch_event;

    for (int i = 0; i < task; i++) {
        dispatch_event = znx_thread_calloc(sizeof(znx_dispatch_event_t));
        dispatch_event->handler = znx_test_client_handler;
        dispatch_event->data = (void *)(intptr_t)i;
        dispatch_event->save_data_1 = (void *)(intptr_t)case_no;
        znx_worker_processor_post_event(dispatch_event);
    }
}


int main()
{
    znx_address_t address;
    bzero(&address, sizeof(znx_address_t));
    if (!znx_address_parser(
        (u_char *)"127.0.0.1:8889", strlen("127.0.0.1:8889"), &address, NULL))
    {
        ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8889");
    }

    znx_proc_init();

    znx_worker_processor_start(2, 2, 0);
    znx_tcp_accepter_start(2);


    znx_listener_t *l = znx_listener_alloc();
    l->listen_addr = "127.0.0.1:8889";
    l->address = address;
    l->backlog = 1024;
    l->nodelay = 1;
    l->reuseaddr = 1;
    l->accept_handler = znx_test_server_accept_handler_1;

    if (!znx_listener_open(l, INVALID_FD)) {
        ZNX_DEF_LOG_FATAL(NULL, "open listener 127.0.0.1:8889 failed");
    }

    znx_rwlock_wlock(&g_listeners->rwlock);
    znx_queue_insert_tail(&g_listeners->head, &l->queue);
    g_listeners->version++;
    znx_rwlock_wunlock(&g_listeners->rwlock);

    znx_tcp_test(1);
    usleep(200000);

    znx_listener_t *l1 = znx_listener_alloc();
    l1->listen_addr = "127.0.0.1:8889";
    l1->address = address;
    l1->backlog = 1024;
    l1->nodelay = 1;
    l1->reuseaddr = 1;
    l1->accept_handler = znx_test_server_accept_handler_2;

    // 如果是同一个listen fd上的变更
    // 1. 将l->socfd ---> l1.
    // 2. 需要设置旧的listener->handover.
    // 3. 从g_listeners将l移除, 在l1上添加g_listeners.
    // 4. 释放l.
    if (!znx_listener_open(l1, l->sockfd)) {
        znx_listener_free(l1);
        ZNX_DEF_LOG_FATAL(NULL, "open listener failed");
    }
    znx_atomic_bool_set_true(&l->handover);

    znx_rwlock_wlock(&g_listeners->rwlock);
    znx_queue_remove(&l->queue);
    znx_queue_insert_tail(&g_listeners->head, &l1->queue);
    g_listeners->version++;
    znx_rwlock_wunlock(&g_listeners->rwlock);
    // 通知tcp acceepter loops update listeners.
    znx_tcp_accepter_notify();
    znx_listener_free(l);

    znx_tcp_test(2);
    usleep(200000);

    // 删除一个listener.
    // 从g_listeners中移除它, 并释放它.
    znx_rwlock_wlock(&g_listeners->rwlock);
    znx_queue_remove(&l1->queue);
    g_listeners->version++;
    znx_rwlock_wunlock(&g_listeners->rwlock);
    // 通知tcp acceepter loops update listeners.
    znx_tcp_accepter_notify();
    znx_listener_free(l1);

    znx_tcp_test(3);
    usleep(200000);

    // 监听新的listener
    bzero(&address, sizeof(znx_address_t));
    if (!znx_address_parser(
        (u_char *)"127.0.0.1:8888", strlen("127.0.0.1:8888"), &address, NULL)) {
        ZNX_DEF_LOG_FATAL(NULL, "parser %s failed", "127.0.0.1:8888");
    }

    znx_listener_t *l2 = znx_listener_alloc();
    l2->listen_addr = "127.0.0.1:8888";
    l2->address = address;
    l2->backlog = 1024;
    l2->nodelay = 1;
    l2->reuseaddr = 1;
    l2->accept_handler = znx_test_server_accept_handler_4;

    if (!znx_listener_open(l2, INVALID_FD)) {
        znx_listener_free(l2);
        ZNX_DEF_LOG_FATAL(NULL, "open listener failed");
    }

    znx_rwlock_wlock(&g_listeners->rwlock);
    znx_queue_insert_tail(&g_listeners->head, &l2->queue);
    g_listeners->version++;
    znx_rwlock_wunlock(&g_listeners->rwlock);
    // 通知tcp acceepter loops update listeners.
    znx_tcp_accepter_notify();

    znx_tcp_test(4);
    usleep(200000);

    znx_proc_shut_down();

    return 0;
}



